package com.ubxtech.mqtt.service.impl;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.util.NumberUtil;
import com.ubxtech.mqtt.config.MqttBeanHolderUtil;
import com.ubxtech.mqtt.config.MqttConstant;
import com.ubxtech.mqtt.service.MqttGateway;
import lombok.extern.slf4j.Slf4j;
import org.springframework.integration.mqtt.outbound.AbstractMqttMessageHandler;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;

import java.util.Map;

/**
 * @author Victor.Xiao
 * @since 2022-11-10 11:23
 **/
@Slf4j
@Component
public class MqttGatewayImpl implements MqttGateway {

    @Override
    public void sendToMqtt(String topic, String payload) {
        Message<String> mqttMessage = MessageBuilder.withPayload(payload).setHeader(MqttHeaders.TOPIC, topic).setHeader(MqttHeaders.RETAINED, false)
                .setHeader(MqttHeaders.QOS, MqttConstant.QOS_1).build();
        getMessageHandler().handleMessage(mqttMessage);
    }

    @Override
    public void sendToMqtt(String topic, int qos, boolean retain, String payload) {
        Message<String> mqttMessage = MessageBuilder.withPayload(payload).setHeader(MqttHeaders.TOPIC, topic).setHeader(MqttHeaders.RETAINED, retain)
                .setHeader(MqttHeaders.QOS, qos).build();
        getMessageHandler().handleMessage(mqttMessage);
    }

    @Override
    public void sendToMqtt(String channel , String topic, String payload) {
        Message<String> mqttMessage = MessageBuilder.withPayload(payload).setHeader(MqttHeaders.TOPIC, topic).setHeader(MqttHeaders.RETAINED, false)
                .setHeader(MqttHeaders.QOS, MqttConstant.QOS_1).build();
        getMessageHandler(channel).handleMessage(mqttMessage);
    }

    @Override
    public void sendToMqtt(String channel , String topic, int qos, boolean retain, String payload) {
        Message<String> mqttMessage = MessageBuilder.withPayload(payload).setHeader(MqttHeaders.TOPIC, topic).setHeader(MqttHeaders.RETAINED, retain)
                .setHeader(MqttHeaders.QOS, qos).build();
        getMessageHandler(channel).handleMessage(mqttMessage);
    }


    /**
     * 发送消息 默认QOS:1 retain:false,MqttHeaders.MESSAGE_EXPIRY_INTERVAL 过期消息
     * @author Victor.Xiao
     * @since 2024-05-31 14:51
     * @param channel channel
     * @param topic topic
     * @param payload payload
     * @param headers key为org.springframework.integration.mqtt.support.MqttHeaders中的值
     */
    @Override
    public void sendToMqtt(String channel , String topic, String payload, Map<String,String> headers) {
        Message<String> mqttMessage = buildMessage(topic, payload, headers);
        getMessageHandler(channel).handleMessage(mqttMessage);
    }

    /**
     * 发送消息 默认QOS:1 retain:false,MqttHeaders.MESSAGE_EXPIRY_INTERVAL 过期消息
     * @author Victor.Xiao
     * @since 2024-05-31 14:51
     * @param topic topic
     * @param payload payload
     * @param headers key为org.springframework.integration.mqtt.support.MqttHeaders中的值
     */
    @Override
    public void sendToMqtt(String topic, String payload, Map<String,String> headers) {
        Message<String> mqttMessage = buildMessage(topic, payload, headers);
        getMessageHandler().handleMessage(mqttMessage);
    }

    private Message<String> buildMessage(String topic, String payload, Map<String,String> headers) {
        MessageBuilder<String> builder = MessageBuilder.withPayload(payload)
                .setHeader(MqttHeaders.TOPIC, topic)
                .setHeader(MqttHeaders.QOS, MqttConstant.QOS_1)
                .setHeader(MqttHeaders.RETAINED, false);

        if (CollectionUtil.isNotEmpty(headers)) {
            headers.forEach(builder::setHeader);
        }

        if (CollectionUtil.isNotEmpty(headers)) {
            headers.forEach((key,value) ->{
                if (MqttHeaders.MESSAGE_EXPIRY_INTERVAL.equalsIgnoreCase(key) && NumberUtil.isNumber(value)) {
                    int second = Integer.parseInt(value) * 1000;
                    long date = System.currentTimeMillis() + second;
                    //设置消息过期时间
                    builder.setExpirationDate(date);
                } else {
                    builder.setHeader(key, value);
                }
            });
        }
//        builder.setExpirationDate(System.currentTimeMillis()+30000);
        return builder.build();
    }



    /**
     * 根据通道获取handler
     * @author Victor.Xiao
     * @since 2022-11-10 11:29
     * @return org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler
     */
    private AbstractMqttMessageHandler getMessageHandler() {
        boolean hasDefault = MqttBeanHolderUtil.hasMessageHandler(MqttConstant.DEFAULT_CHANNEL_NAME);
        if (hasDefault) {
            return getMessageHandler(MqttConstant.DEFAULT_CHANNEL_NAME);
        }
        AbstractMqttMessageHandler handler = MqttBeanHolderUtil.getDefaultMessageHandler();
        if (handler == null) {
            log.error("没有MessageHandler");
            throw new IllegalArgumentException("没有MessageHandler");
        }
        return handler;
    }

    /**
     * 根据通道获取handler
     * @author Victor.Xiao
     * @since 2022-11-10 11:29
     * @param channelName channelName
     * @return org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler
    */
    private AbstractMqttMessageHandler getMessageHandler(String channelName) {
        AbstractMqttMessageHandler handler = MqttBeanHolderUtil.getMessageHandler(channelName);
        if (handler == null) {
            log.error("没有找到通道{}的MessageHandler", channelName);
            throw new IllegalArgumentException("没有找到通道" + channelName + "的MessageHandler");
        }
        return handler;
    }

}
